package cn.tannn;


import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.SqlDialect;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class FlinkCDCWitchSQL {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        Configuration configuration = new Configuration();
//        configuration.setString(RestOptions.BIND_PORT, "8081-8089");
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);
//        env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("file:///F://project//java//flink-cdc-demo-1//doc"));
        // 每隔1000 ms进行启动一个检查点【设置checkpoint的周期】
        env.enableCheckpointing(3000);
        // 高级选项：
        // 设置模式为exactly-once （这是默认值）
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        // 确保检查点之间有至少500 ms的间隔【checkpoint最小间隔】
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1500);
        // 检查点必须在一分钟内完成，或者被丢弃【checkpoint的超时时间】
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        // 同一时间只允许进行一个检查点
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        // 表示一旦Flink处理程序被cancel后，会保留Checkpoint数据，以便根据实际需要恢复到指定的Checkpoint【详细解释见备注】
        //ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION:表示一旦Flink处理程序被cancel后，会保留Checkpoint数据，以便根据实际需要恢复到指定的Checkpoint
        //ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION: 表示一旦Flink处理程序被cancel后，会删除Checkpoint数据，只有job执行失败的时候才会保存checkpoint
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.setParallelism(1);

        EnvironmentSettings Settings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, Settings);
        tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);
        // 数据源表
        String sourceDDL =  "CREATE TABLE mysql_binlog1 ( " +
                                " id INT NOT NULL, " +
                                " tm_name STRING, " +
                                " logo_url STRING, " +
                                " primary key (id) not enforced " +
                                ") WITH ( " +
                                " 'connector' = 'mysql-cdc', " +
                                " 'hostname' = '192.168.0.99', " +
                                " 'port' = '3306', " +
                                " 'username' = 'root', " +
                                " 'password' = 'root', " +
                                " 'database-name' = 'flink', " +
                                " 'table-name' = 'base_trademark' " +
//                                " , 'scan.startup.mode' = 'latest-offset' " + // 默认全量加增量
                                ")  ";
        // 输出目标表 （一样的字段）
        String sinkDDL =
                "CREATE TABLE test_cdc_sink1 ( " +
                        " id INT, " +
                        " tm_name STRING, " +
                        " logo_url STRING, " +
                        " primary key (id) not enforced " +
                        ") WITH ( " +
                        " 'connector' = 'jdbc', " +
                        " 'driver' = 'com.mysql.cj.jdbc.Driver', " +
                        " 'url' = 'jdbc:mysql://192.168.0.51:3316/test?serverTimezone=Asia/Shanghai&useSSL=false'," +
                        " 'username' = 'root', " +
                        " 'password' = 'root', " +
                        " 'table-name' = 'base_trademark' " +
                        ")";
        // 输出目标表 （不一样的字段  没有主键 更新和删除会出问题，实在没有主键的情况下必须设置唯一索引）
        String sinkDDL2 =
                "CREATE TABLE test_cdc_sink2 ( " +
                        " name STRING, " +
                        " logo STRING, " +
                        " primary key (name) not enforced " +
                        ") WITH ( " +
                        " 'connector' = 'jdbc', " +
                        " 'driver' = 'com.mysql.cj.jdbc.Driver', " +
                        " 'url' = 'jdbc:mysql://127.0.0.1:3316/test?serverTimezone=Asia/Shanghai&useSSL=false'," +
                        " 'username' = 'root', " +
                        " 'password' = 'root', " +
                        " 'table-name' = 'test' " +
                        ")";
        // 简单的聚合处理
        String transformDmlSQL =  "insert into test_cdc_sink1 select * from mysql_binlog1";
        String transformDmlSQL2 =  "insert into test_cdc_sink2 select tm_name as name , logo_url as logo from mysql_binlog";

        TableResult sourceSQL = tableEnv.executeSql(sourceDDL);
        TableResult sinkSQL = tableEnv.executeSql(sinkDDL);
        TableResult transformSQL = tableEnv.executeSql(transformDmlSQL);
//        tableEnv.executeSql(sinkDDL2);
//        tableEnv.executeSql(transformDmlSQL2);
        // 下面的不注释上传到集群会报错
//        sourceSQL.print();
//        transformSQL.print();
//        sinkSQL.print();
//        env.execute("mysql-sql-mysql-cdc");
    }

}
